package com.im.service.group.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.im.common.constant.Constants;
import com.im.common.enums.command.GroupEventCommand;
import com.im.common.model.message.GroupChatMessageContent;
import com.im.common.model.message.MessageReadedContent;
import com.im.service.group.service.GroupMessageService;
import com.im.service.message.service.MessageSyncService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 群聊天消息接收操作类
 */
@Slf4j
@Component
public class GroupChatOperateReceiver {

    @Autowired
    private GroupMessageService groupMessageService; // 自动注入群消息服务
    @Autowired
    private MessageSyncService messageSyncService; // 自动注入消息同步服务

    // 监听RabbitMQ队列，接收消息
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = Constants.RabbitConstants.Im2GroupService, durable = "true"), // 定义队列名称及持久化
                    exchange = @Exchange(value = Constants.RabbitConstants.Im2GroupService, durable = "true") // 定义交换机名称及持久化
            ), concurrency = "1" // 设置并发数为1
    )
    public void onChatMessage(@Payload Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        String msg = new String(message.getBody(), "utf-8"); // 将消息体转换为字符串
        log.info("CHAT MSG FORM QUEUE ::: {}", msg); // 打印日志
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG); // 获取消息的deliveryTag
        try {
            JSONObject jsonObject = JSON.parseObject(msg); // 将字符串解析为JSON对象
            Integer command = jsonObject.getInteger("command"); // 获取消息中的command字段
            if (command.equals(GroupEventCommand.MSG_GROUP.getCommand())) {
                // 处理消息
                GroupChatMessageContent messageContent = jsonObject.toJavaObject(GroupChatMessageContent.class);
                groupMessageService.process(messageContent);
            } else if (command.equals(GroupEventCommand.MSG_GROUP_READED.getCommand())) {
                // 处理已读消息
                MessageReadedContent messageReaded = JSON.parseObject(msg, new TypeReference<MessageReadedContent>() {
                }.getType());
                messageSyncService.groupReadMark(messageReaded);
            }
            channel.basicAck(deliveryTag, false); // 手动确认消息已被消费
        } catch (Exception e) {
            log.error("处理消息出现异常：{}", e.getMessage()); // 打印异常信息
            log.error("RMQ_CHAT_TRAN_ERROR", e); // 打印异常堆栈信息
            log.error("NACK_MSG:{}", msg); // 打印未被消费的消息
            // 第一个false表示不批量拒绝，第二个false表示不重回队列
            channel.basicNack(deliveryTag, false, false); // 手动拒绝消息并且不重回队列
        }
    }
}
